#include"pm_ehash.h"
#include <iostream>
using namespace std;
PmEHash::PmEHash() {//ok
	char path[128] = PM_EHASH_DIRECTORY;
	strcat(path,META_NAME);
	if (access(path, F_OK) != -1){
		recover();
	}
	else{
		/* 初始化metadata */
		int is_pmem;
		size_t map_len;

		metadata = (ehash_metadata*)pmem_map_file(path, sizeof(ehash_metadata), PMEM_FILE_CREATE, 0777, &map_len, &is_pmem);

		if ( !pmem_is_pmem(metadata, map_len) ){
			//cout << "fail in PmEHash: can not create metadata" << endl;
		}

		metadata->max_file_id = 1;
		metadata->catalog_size = 2;
		metadata->global_depth = 1;

		/* 初始化catalog, 由于只需要使用到bucket_vitrual_address, 我们只创建这个变量的文件存储，我们也只使用这个变量,但是在存储时要存bucket_address */

		catalog.buckets_virtual_address = new pm_bucket* [2];

		page_list = new data_page* [metadata->catalog_size];
		page_list_len = metadata->catalog_size;
		allocNewPage();
		pm_address new_address;
		catalog.buckets_virtual_address[0] = (pm_bucket*)getFreeSlot(new_address);
		catalog.buckets_virtual_address[1] = (pm_bucket*)getFreeSlot(new_address);
	}
}
/**
 * @description: persist and munmap all data in NVM
 * @param NULL
 * @return: NULL
 */

PmEHash::~PmEHash() {
	/* 将catalog中的buckets_virtual_address 对应的bucket_address存入NVM并关闭映射 */
	char cpath[128] = PM_EHASH_DIRECTORY;
	strcat(cpath, CATALOG_NAME);
	int cis_pmem;
	size_t cmap_len;
	catalog.buckets_pm_address = (pm_address *)pmem_map_file(cpath, metadata->catalog_size*sizeof(pm_address), PMEM_FILE_CREATE,0777, &cmap_len, &cis_pmem);

	if ( !pmem_is_pmem(catalog.buckets_pm_address, cmap_len) ) {
		//cout << "fail in PmEHash: can not create catalog" << endl;
	}

	for (int i = 0; i < metadata->catalog_size; ++i){

		catalog.buckets_pm_address[i].fileId = vAddr2pmAddr[catalog.buckets_virtual_address[i]].fileId;
		catalog.buckets_pm_address[i].offset = vAddr2pmAddr[catalog.buckets_virtual_address[i]].offset;
		////cout << "~PmEhash: pm_address " << catalog.buckets_pm_address[i].fileId <<" " << catalog.buckets_pm_address[i].offset << endl;
	}


	pmem_persist( catalog.buckets_pm_address, (metadata->catalog_size)*( sizeof(pm_address) ) );
	pmem_unmap(catalog.buckets_pm_address,  (metadata->catalog_size)*( sizeof(pm_address) ) );

	/* 将所有的data_page的数据写会NVM并关闭映射 */

	for (int i = 0; i < metadata->max_file_id - 1; ++i){
		pmem_persist(page_list[i], sizeof(data_page));
		pmem_unmap(page_list[i], sizeof(data_page));
	}

	/* 将metadata 存入NVM并关闭映射 */
	pmem_persist(metadata, sizeof(ehash_metadata));
	pmem_unmap(metadata, sizeof(ehash_metadata));
}


int PmEHash::insert(kv kv_pair) {
    uint64_t temp;
    if(search(kv_pair.key,temp) == 0){
    	//cout << kv_pair.key << "is in bucket" << endl;
    	return -1;
    }

    pm_bucket* bucket = getFreeBucket(kv_pair.key);

    kv* freeSlot = getFreeKvSlot(bucket);
    freeSlot->key = kv_pair.key;
    freeSlot->value = kv_pair.value;
    return 0;

}
int PmEHash::remove(uint64_t key) { //ok

	/*如果不存在这条记录，返回-1*/
	uint64_t temp;
    if(search(key,temp) == -1)
		return -1;

	/*存在这条记录，删除它*/
	pm_bucket* bucket;

	int bucket_id = hashFunc(key);
	bucket = catalog.buckets_virtual_address[bucket_id];

	for(int i = 0;i < BUCKET_SLOT_NUM; i++){
		if(bucket->slot[i].key == key && bucket->bitmap[i] == 1){
			bucket->bitmap[i] = 0;
		}
	}

	bool is_empty = true;

	for (int i = 0; i < BUCKET_SLOT_NUM; ++i){
		if (bucket->bitmap[i] == 1){
			is_empty = false;
			break;
		}
	}

	if(is_empty){
		mergeBucket(bucket_id);
	}

	return 0;
}

int PmEHash::update(kv kv_pair) {
    uint64_t temp;
    if(search(kv_pair.key,temp) == -1)
		return -1;
	pm_bucket* bucket;

	if (kv_pair.value == temp){
		return 0;
	}

	int bucket_id = hashFunc(kv_pair.key);
	bucket = catalog.buckets_virtual_address[bucket_id];
	for(int i = 0;i < BUCKET_SLOT_NUM;i++){
		if(bucket->slot[i].key == kv_pair.key){
			bucket->slot[i].value = kv_pair.value;
			return 0;
		}
	}
	return -1;
}

int PmEHash::search(uint64_t key, uint64_t& return_val) {
    uint64_t bucket_id = hashFunc(key);
    pm_bucket* bucket = catalog.buckets_virtual_address[bucket_id];
	if (bucket == NULL) {
		return -1;
	}
    for(int i = 0;i< BUCKET_SLOT_NUM; i++){
    	if (((bucket->bitmap[i]) == 1) && (bucket->slot[i].key == key)){
            return_val = bucket->slot[i].value;
    		return 0;
    	}
    }

    return -1;
}


uint64_t PmEHash::hashFunc(uint64_t key) {
    uint64_t level = 2;
    for(int i = 1;i < metadata->global_depth;i++){
		level *= 2;
    }
    key = key % level;
    return key;
}

pm_bucket* PmEHash::getFreeBucket(uint64_t key) {
    uint64_t bucket_id = hashFunc(key);
    pm_bucket* bucket = catalog.buckets_virtual_address[bucket_id];
    if (bucket == NULL){
    	cout << "fail in get FreeBucket: a bucket is NULL bucket_id == "<< bucket_id << endl;
    }

    for(int i = 0;i< BUCKET_SLOT_NUM;i++){

        if(bucket->bitmap[i] == 0){
        	return bucket;
        }
    }

	uint64_t level = 1;
	for (int i = 0; i < bucket->local_depth; ++i){
		level *= 2;
	}

	if (bucket_id >= level){
		int past_bucket_id = bucket_id -level;
		pm_bucket* past_bucket = catalog.buckets_virtual_address[bucket_id - level];
		while (past_bucket == bucket){
			past_bucket_id = past_bucket_id - level;
			if (past_bucket_id < 0) break;
			past_bucket = catalog.buckets_virtual_address[past_bucket_id];
		}
		splitBucket(past_bucket_id + level);
	}
	else{
		splitBucket(bucket_id);
	}

    return getFreeBucket(key);
}

kv* PmEHash::getFreeKvSlot(pm_bucket* bucket) {
    for(int i = 0;i< BUCKET_SLOT_NUM;i++){
        if( bucket->bitmap[i] == 0){
        	bucket->bitmap[i] = 1;
        	return &(bucket->slot[i]);
        }
    }

    return NULL;
}


void PmEHash::splitBucket(uint64_t bucket_id) {
    pm_bucket* bucket = catalog.buckets_virtual_address[bucket_id];
    pm_bucket* new_bucket = NULL;
    pm_address new_bucket_address;
    new_bucket = (pm_bucket *) getFreeSlot(new_bucket_address);
    new_bucket->local_depth = bucket->local_depth + 1;
	if ((bucket->local_depth+1) > metadata->global_depth){
		extendCatalog();
	}
	int level = 1;
	for (int i = 0; i < bucket->local_depth; ++i){
		level *= 2;
	}
	int new_bucket_id = level + bucket_id;
	catalog.buckets_virtual_address[new_bucket_id] = new_bucket;
	level *= 2;
	for (int i = new_bucket_id + level; i < metadata->catalog_size; i += level){
		catalog.buckets_virtual_address[i] = new_bucket;
	}

	bucket->local_depth += 1;
    for (int i = 0; i < BUCKET_SLOT_NUM; ++i){
    	uint64_t key = bucket->slot[i].key;
    	if ((key % level) == new_bucket_id){
    		kv * free_kv = getFreeKvSlot(new_bucket);
    		free_kv->key = bucket->slot[i].key;
    		free_kv->value = bucket->slot[i].value;
    		bucket->bitmap[i] = 0;
    	}
    }
}

void PmEHash::mergeBucket(uint64_t bucket_id) {
	if (metadata->catalog_size == 2) return;
	pm_bucket * old_bucket = catalog.buckets_virtual_address[bucket_id];
	uint64_t level = 1;
	for (int i = 1; i < old_bucket->local_depth; ++i){
		level *= 2;
	}
	bool real_merge = true;
	int real_bucket_id = bucket_id;
	pm_bucket * real_bucket = catalog.buckets_virtual_address[bucket_id];
	while (old_bucket == real_bucket){
		real_bucket_id = real_bucket_id - level;
		if (real_bucket_id < 0) break;
		real_bucket = catalog.buckets_virtual_address[real_bucket_id];
	}
	real_bucket_id += level;
	/*如果要删除的桶位于哈希树的右节点，只有在要删除的桶和左节点深度相同时删除桶，否则不回收桶*/
	if (real_bucket_id >= level){
		int past_bucket_id = real_bucket_id - level;
		if (catalog.buckets_virtual_address[past_bucket_id]->local_depth == catalog.buckets_virtual_address[real_bucket_id]->local_depth){
			level *= 2;
			for (int i = real_bucket_id; i < metadata->catalog_size; i += level){
				catalog.buckets_virtual_address[i] = catalog.buckets_virtual_address[past_bucket_id];
			}
			catalog.buckets_virtual_address[real_bucket_id]->local_depth -= 1;
		}
		else
            real_merge = false;

	}
	/*如果要删除的桶位于哈希树的左节点，只有在要删除的桶和右节点深度相同时删除桶，否则不回收桶*/
	else{
		int next_bucket_id = real_bucket_id + level;
		if (catalog.buckets_virtual_address[next_bucket_id]->local_depth == catalog.buckets_virtual_address[real_bucket_id]->local_depth){
			level *= 2;
			for (int i = real_bucket_id; i < metadata->catalog_size; i += level){
				catalog.buckets_virtual_address[i] = catalog.buckets_virtual_address[next_bucket_id];
			}
			catalog.buckets_virtual_address[real_bucket_id]->local_depth -= 1;
		}
		else real_merge = false;
	}

	if (real_merge){
		for (int i = 0; i < BUCKET_SLOT_NUM; ++i){
			old_bucket->bitmap[i] = 0;
		}
		old_bucket->local_depth = 1;
		free_list.push(old_bucket);
	}
}


void PmEHash::extendCatalog() {
    int cis_pmem;
	size_t cmap_len;
	char cpath[128] = PM_EHASH_DIRECTORY;
	strcat(cpath, CATALOG_NAME);

	/* 保存原来的catalog.buckets_virtual_address到t_catalog中， 并删除原来的catalog.buckets_virtual_address */
	pm_bucket* t_catalog[metadata->catalog_size];

	for (int i = 0; i < metadata->catalog_size; ++i){
		t_catalog[i] = catalog.buckets_virtual_address[i];
	}
    /* 新建一个catalog.buckets_virtual_address，大小为原来的两倍，将t_catalog复制到里面 */
	catalog.buckets_virtual_address = new pm_bucket* [2*(metadata->catalog_size)];

	for (int i = 0; i < metadata->catalog_size; ++i){
		catalog.buckets_virtual_address[i] = t_catalog[i];
		catalog.buckets_virtual_address[i+metadata->catalog_size] = NULL;
	}

	for (int i = 0; i < metadata->catalog_size; ++i){
		uint64_t level = 1;
		for (int j = 0; j < catalog.buckets_virtual_address[i]->local_depth; ++j){
			level *= 2;
		}
		for (int j = i + level; j < (metadata->catalog_size * 2); j += level){
			if (catalog.buckets_virtual_address[j] != NULL)
                break;
			catalog.buckets_virtual_address[j] = catalog.buckets_virtual_address[i];
		}
	}

	metadata->catalog_size *= 2;
	metadata->global_depth++;
}

void* PmEHash::getFreeSlot(pm_address& new_address) {
	if (free_list.empty()){
		allocNewPage();
	}
	/* 从free_list中取出一个空的bucket， 并将对应数据页中bitmap相应位置置为1 */
	pm_bucket * res = free_list.front();
	res->local_depth = 1;
	free_list.pop();
	new_address.fileId = 1;
	new_address.offset = 1;
	if (vAddr2pmAddr.count(res) > 0){
		new_address.fileId = vAddr2pmAddr[res].fileId;
		new_address.offset = vAddr2pmAddr[res].offset;
    }
	page_list[new_address.fileId - 1]->bitmap[new_address.offset] = 1;
	return res;
}

void PmEHash::allocNewPage() {//ok
	uint64_t fileID;
	size_t map_len;
	/* 根据metadata的第一个元素的max_file_id 创建一个新的页面表，并将页面表中的bucket放入free_list中，以及更新vAddr2pmAddr pmAddr2vAddr*/
	fileID = metadata->max_file_id;
	data_page* tdata = newPage(fileID, map_len);
	page_list[fileID - 1] = tdata;
	if (fileID == page_list_len){
		data_page ** t_page = new data_page* [page_list_len * 2];
		for (int i = 0; i < page_list_len; ++i){
			t_page[i] =  page_list[i];
			t_page[i+page_list_len] = NULL;
		}
		page_list_len *= 2;
		delete [] page_list;
		page_list = t_page;
	}
	for (int i = 0; i < DATA_PAGE_SLOT_NUM; ++i){
		pm_address t_RID;
		t_RID.fileId = fileID;
		t_RID.offset = i;
		pm_bucket* t_address = &(tdata->slot[i]);
		free_list.push(t_address);
        vAddr2pmAddr.insert(make_pair(t_address, t_RID));
	}

	metadata->max_file_id += 1;
	//cout << "create page " << metadata->max_file_id -1 << "success" << endl;
}

/**
 * @description: 读取旧数据文件重新载入哈希，恢复哈希关闭前的状态
 * @param NULL
 * @return: NULL
 */
void PmEHash::recover() {
	/*从硬盘中读回 metadate*/
	string metadata_path = DATA;
	metadata_path += META_NAME;
    size_t metadata_len;
	int is_pmem;

	if ( (metadata = (ehash_metadata*)pmem_map_file(metadata_path.c_str(), sizeof(ehash_metadata) , PMEM_FILE_CREATE,
			0666, &metadata_len, &is_pmem)) == NULL ) {
        perror("pmem_map_file");
        exit(1);
        }

	/*新建page_list记录各个页面表的物理地址*/
	page_list_len = 2;
	while (page_list_len < metadata->max_file_id) page_list_len *= 2;
	page_list = new data_page* [page_list_len];
	for (int i = 0; i < page_list_len; ++i){
		page_list[i] = NULL;
	}
	/*将数据页进行映射*/
	mapAllPage();
	/*从硬盘中读回catalog*/
	string catalog_path = DATA;
	catalog_path += CATALOG_NAME;
	size_t catalog_len;

	if ((catalog.buckets_pm_address = (pm_address *)pmem_map_file(catalog_path.c_str(), metadata->catalog_size * sizeof(pm_address), PMEM_FILE_CREATE,
			0666, &catalog_len, &is_pmem)) == NULL) {
        perror("pmem_map_file");
        exit(1);
    }

	catalog.buckets_virtual_address = new pm_bucket*[metadata->catalog_size];
	for (int i = 0; i < metadata->catalog_size; ++i){
		int fileId = catalog.buckets_pm_address[i].fileId;
		int offset = catalog.buckets_pm_address[i].offset;
		catalog.buckets_virtual_address[i] = &(page_list[fileId - 1]->slot[offset]);
	}
}


void PmEHash::mapAllPage() {
	int is_pmem;
	for (uint64_t i = 1; i < (metadata->max_file_id); ++i) {
		char* num = new char[50];
		string path = DATA;
		sprintf(num, "%ld", i);
		string page_path = path + num;
		size_t page_len;
		if ((page_list[i-1] = (data_page*)pmem_map_file(page_path.c_str(), sizeof(data_page), PMEM_FILE_CREATE,
			0666, &page_len, &is_pmem)) == NULL) {
            perror("pmem_map_file");
            exit(1);
        }

		for (int j = 0; j < DATA_PAGE_SLOT_NUM; ++j) {
            pm_address t_RID;
		    t_RID.fileId = i;
		    t_RID.offset = j;
		    pm_bucket* bucket = &(page_list[i-1]->slot[j]);
	    	vAddr2pmAddr.insert(make_pair(bucket, t_RID));
			if (page_list[i-1]->bitmap[j] == 0){
				free_list.push(bucket);
			}
        }
	}
}

void PmEHash::selfDestory() {
	/*删除catalog*/
	string cpath = DATA;
	cpath += CATALOG_NAME;
	if (access(cpath.c_str(), F_OK) != -1){
		if (::remove(cpath.c_str()) == 0) {
        }
		else{
            cout << "fail in selfDestory: delete catalog"<< endl;
		}
	}

	/*删除metadata*/
	pmem_unmap(metadata, sizeof(ehash_metadata));
	string mpath = DATA;
	mpath += META_NAME;

	if (::remove(mpath.c_str()) == 0) {}

	else
		cout << "fail in selfDestory: delete metadata"<< endl;
}

